package terra.network.handler;

import java.util.concurrent.ConcurrentLinkedQueue;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

import terra.BootStrap;
import terra.network.Server;
import terra.network.packet.Packet;


public abstract class ClientHandler extends SimpleChannelUpstreamHandler {
	
	private Channel channel;
	
	protected ConcurrentLinkedQueue<Packet> readQueue = new ConcurrentLinkedQueue<Packet>();
	protected ConcurrentLinkedQueue<Packet> sendQueue = new ConcurrentLinkedQueue<Packet>();
	protected ConcurrentLinkedQueue<Packet> sendHighQueue = new ConcurrentLinkedQueue<Packet>();
	protected ConcurrentLinkedQueue<Packet> sendLowQueue = new ConcurrentLinkedQueue<Packet>();
	
	protected Thread readPacketTask = new Thread(new ReadHandler());
	protected Thread sendPacketTask = new Thread(new SendHandler());

	
	public abstract void handlePacket(Packet p);

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
    	if (e.getMessage() instanceof Packet) {
    		readQueue.add((Packet) e.getMessage());
    	}
    }

    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
    	//System.out.println("channelConnected " + ctx.getChannel());
    	setChannel(ctx.getChannel());
    	readPacketTask.start();
    	sendPacketTask.setPriority(Thread.MAX_PRIORITY-3);
    	sendPacketTask.start();

    	if (BootStrap.mode == BootStrap.SERVER_MODE)
    		Server.getInstance().all.add(ctx.getChannel());
    }
    
    @Override
    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
    	//System.out.println("channelDisconnected " + ctx.getChannel());
    	cancelTasks();
    	getChannel().close();
    	if (BootStrap.mode == BootStrap.SERVER_MODE)
    		Server.getInstance().all.remove(ctx.getChannel());
    }
    
	private void cancelTasks() {
		readQueue.clear();
    	//readQueue = null;
    	sendQueue.clear();
    	sendLowQueue.clear();
    	sendHighQueue.clear();
    	//sendQueue = null;
    	readPacketTask.interrupt();
    	sendPacketTask.interrupt();
	}

	public void send(Packet p) {
		if (p != null){
			if (p.getPriority() == Packet.PRIOR_HIGH)
				sendHighQueue.add(p);
			else if  (p.getPriority() == Packet.PRIOR_LOW)
				sendLowQueue.add(p);
			else
				sendQueue.add(p);
		}
	}

    public Channel getChannel() {
		return channel;
	}

	public void setChannel(Channel channel) {
		this.channel = channel;
	}
	
	public boolean readQueueIsEmpty() {
		return readQueue.isEmpty();
	}
	
	public Packet getNextReadablePacket() {
		return readQueue.poll();
	}
	
	public boolean sendQueueIsEmpty() {
		return sendQueue.isEmpty();
	}
	
	public Packet getNextWritablePacket() {
		return sendQueue.poll();
	}
	
	class ReadHandler implements Runnable {
		@Override
		public void run() {
			while(getChannel().isConnected()){
				while(!readQueueIsEmpty())
					handlePacket(getNextReadablePacket());
				try {
					Thread.sleep(0,50);
				} catch (InterruptedException e) {
					//e.printStackTrace();
				}
			}
		}
    }
	
	class SendHandler implements Runnable {
		@Override
		public void run() {
			while(getChannel().isConnected()){
				if (!sendHighQueue.isEmpty()){
					for(int i = 0; i < 10;i++){
						if (sendHighQueue.isEmpty())
							break;
						getChannel().write(sendHighQueue.poll());
					}
				}
				if (!sendQueue.isEmpty())
					for(int i = 0; i < 5;i++){
						if (sendQueue.isEmpty())
							break;
						getChannel().write(sendQueue.poll());
					}
				if (!sendLowQueue.isEmpty())
					getChannel().write(sendLowQueue.poll());
				
//				while(!sendQueueIsEmpty())
//					getChannel().write(getNextWritablePacket());
				try {
					Thread.sleep(0,50);
				} catch (InterruptedException e) {
					//e.printStackTrace();
				}	
			}
		}
    }
}
